package net.tp.parallel;

import static net.tp.util.Utils.*;
import net.tp.algo.sort.MergeSort;
import net.tp.struct.bst.*;
import java.util.concurrent.*;
import java.util.*;

/**
 * Parallel Merge Sort.
 *
 * <p>Merge sort algorithm basically divides and merges data. It's straightforward to apply multithread for divisions. For merging, we divide the data in such a way that can be done in parallel. Assuming that we want to merge 2 arrays a[r1:s1] and a[r2:s2] into this array b[r3:s3], let m be the mid point of (r1,s1), let x = a[m] be the median of a[r1:s1], do binary search on a[r2:s2] to find index n such that a[r2:n] ≤ x and a[n+1:s2] ≥ x. So now we can merge a[r1:m] with a[r2:m] and merge a[m+1:s1] with a[n+1:s2] independently.</p>
 *
 * @author Trung Phan
 * @see <p>Multithreaded merge sort in Introduction to Algorithms (Cormen), Third Edition</p>
 */
public class ParallelMergeSort {

    private final static ForkJoinPool forkJoinPool = new ForkJoinPool();

    private final static int THRESHOLD = 2; // threshold for sequential processing. Should be large.

    public static void sort(int[] a) {
        sort(a, 0, a.length - 1);
    }

    public static void sort(int[] a, int lo, int hi) {
        int[] b = new int[a.length];
        forkJoinPool.invoke(makeRecursiveAction( ()-> mergeSort(a, b, lo, hi)));
    }

    private static void mergeSort(int[] a, int[] b, int lo, int hi) {
        if (hi - lo < THRESHOLD) {
            MergeSort.sort(a, lo, hi);
        }
        else {
            int mid = (lo+hi) >>> 1;
            RecursiveAction action = makeRecursiveAction( ()-> mergeSort(a, b, lo, mid));
            action.fork();
            mergeSort(a, b, mid+1, hi);
            action.join();

            merge(a, b, lo, mid, mid+1, hi, lo, hi);
            parallelCopy(b, lo, a, lo, hi-lo+1);
        }
    }

    private static void merge(int[] a, int[] b, int r1, int s1, int r2, int s2, int r3, int s3) {

        assert s3-r3+1 == (s2-r2+1) + (s1-r1+1);

        if (s3-r3+1 <= THRESHOLD)
            seqMerge(a, b, r1, s1, r2, s2, r3, s3);
        else
            parMerge(a, b, r1, s1, r2, s2, r3, s3);
    }

    private static void seqMerge(int[] a, int[] b, int r1, int s1, int r2, int s2, int r3, int s3) {

        int left = r1, right = r2;
        for (int i = r3; i <= s3; i++) {
            if (left > s1) {
                System.arraycopy(a, right, b, i, s3-i+1);
                break;
            }
            else if (right > s2) {
                System.arraycopy(a, left, b, i, s3-i+1);
                break;
            }
            else if (a[left] <= a[right]) b[i] = a[left++];
            else b[i] = a[right++];
        }
    }
    
    private static void parMerge(int[] a, int[] b, int r1, int s1, int r2, int s2, int r3, int s3) {

        final int p, q;
        if (s1 - r1 > s2 - r2) {
            p = (r1 + s1) >>> 1;
            q = BinarySearch.findPredecessor(a, r2, s2+1, a[p]);
        }
        else {
            q = (r2 + s2) >>> 1;
            p = BinarySearch.findPredecessor(a, r1, s1+1, a[q]);
        }

        int n1 = p-r1+1 + q-r2+1;

        RecursiveAction action = makeRecursiveAction( () -> merge(a, b, r1, p, r2, q, r3, r3+n1-1));
        action.fork();
        merge(a, b, p+1, s1, q+1, s2, r3+n1, s3);
        action.join();
    }
}
